package ru.hh.team02.chat.messaging;

import ru.hh.team02.chat.client.Client;
import ru.hh.team02.chat.utils.BatchBlockingQueue;
import ru.hh.team02.chat.utils.SimpleBlockingQueue;

/**
 * Class responsible for sending messages. Contains a message queue and a sender
 * threads. Threads take messages from the queue and send them.
 */
public class SendingQueue {
    private final BatchBlockingQueue<Message> blockingQueue = new SimpleBlockingQueue<Message>();
    private volatile boolean terminating = false;

    // As we have only one recipient per queue, we may keep it here, not inside
    // the Message class
    private Client recepient;

    /**
     * Create a new SendingBlockingQueue object
     * 
     * @param numberOfThreads
     *            Number of sending threads.
     * @param recepient
     *            Client to receive sending messages
     */
    public SendingQueue(int numberOfThreads, Client recepient) {
        this.recepient = recepient;

        // Creating sending threads and starting them
        for (int i = 0; i < numberOfThreads; i++) {
            Thread sender = new Thread(new Sender());
            sender.start();
        }
    }

    /**
     * Class defining a Sender implementing Runnable interface, which gets a
     * message from a queue and sends it to the recipient.
     * 
     */
    class Sender implements Runnable {
        public void run() {
            while (true) {
                try {
                    // Get hold of a message from the queue
                    // Will throw InterruptedException if in terminating mode
                    Message message = getMessage();

                    // Send the message in this thread
                    recepient.sendMessageToClientComputer(message);
                } catch (InterruptedException e) {
                    // This is the message to stop thread according to the
                    // contract.
                    return;
                }
            }
        }
    }

    /**
     * Get a message from the queue.
     * 
     * @return message
     * @throws InterruptedException
     *             if queue is empty and in terminating mode.
     */
    private Message getMessage() throws InterruptedException {
        if (terminating) {
            throw new InterruptedException("Terminating sending thread.");
        }
        return blockingQueue.get();
    }

    /**
     * Add message to the queue
     * 
     * @param message
     *            message to add to the queue for sending
     * @exception IllegalStateException
     *                Thrown when the queue is closed
     */
    public boolean add(Message message) {
        if (terminating) {
            return false;
        }
        blockingQueue.add(message);
        return true;
    }

    /**
     * Terminate the sending threads as soon as possible. The messages may get
     * stuck in the queue unsent. This is made this way because we will only be
     * using this method if the client has disconnected, and the messages can't
     * be sent to them anyway.
     * 
     */
    public void terminate() {
        // put SendingQueue to terminating mode
        terminating = true;

        // Put BlockingQueue to terminating mode
        blockingQueue.terminate();

        // Waking up all the sending threads blocked
        // blockingQueue waiting in get method for a new message
        // They will throw interrupted exception if the queue is empty
        // Or they will return a new message, then on trying
        // to get a new message via getMessage of this class will throw
        // InterruptedException
        synchronized (blockingQueue) {
            blockingQueue.notifyAll();
        }

    }
}
